package com.gitee.sparta.core.process.feature;

import com.gitee.sparta.common.SpartaConstant;
import com.gitee.sparta.common.helpers.JacksonHelper;
import com.gitee.sparta.common.helpers.RedisKeyHelper;
import com.gitee.sparta.core.bean.SpartaChannel;
import com.gitee.sparta.core.bean.metadata.Metadata;
import com.gitee.sparta.core.process.operate.DenyTabOperate;
import com.gitee.sparta.core.process.operate.ReadableTabOperate;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.extern.slf4j.Slf4j;

/**
 * 消息 ack 特性
 *
 * @author zyred
 * @since v 0.0.1
 */
@Slf4j
public class SpartaAcknowledgeFeature<V> {

    /** ack 标记 **/
    public static ThreadLocal<Integer> ACK_TAG = null;

    private final SpartaCoreProperties properties;
    private final DenyTabOperate<V> denyTabOperate;
    private final ReadableTabOperate<V> readableTabOperate;
    private final SpartaDistributedRenewFeature<V> renewFeature;

    public SpartaAcknowledgeFeature(SpartaCoreProperties properties,
                                    DenyTabOperate<V> denyTabOperate,
                                    ReadableTabOperate<V> readableTabOperate,
                                    SpartaDistributedRenewFeature<V> renewFeature) {
        this.properties = properties;
        this.renewFeature = renewFeature;
        this.denyTabOperate = denyTabOperate;
        this.readableTabOperate = readableTabOperate;

        // 根据配置文件初始化是否自动 ack
        ACK_TAG = ThreadLocal.withInitial(() -> properties.isEnableAutoAck()
                ? SpartaConstant.AUTO_ACK
                : SpartaConstant.MANUAL_ACK);
    }


    /**
     * 开始执行重试， 此重试是分布式的重试，防止单机失败的情况出现
     *
     * @param fullTopic         完整的 topic：topic:metaId
     * @param listenTopic       @SpartaListener 内中 topics 描述
     * @param metadata          被深拷贝的 body 对象，防止用户在回调方法中进行修改内容
     * @param channel           用户 ack 等操作的运用
     */
    public void doProcessAcknowledge(final String fullTopic, final String listenTopic, Metadata<V> metadata, SpartaChannel channel) {

        final int tag = SpartaAcknowledgeFeature.removeAck();
        final boolean acknowledge = (tag & SpartaConstant.DO_ACK) == SpartaConstant.DO_ACK,
                renew = (tag & SpartaConstant.DO_RENEW) == SpartaConstant.DO_RENEW;

        // 确认
        if (acknowledge) {
            this.doAcknowledge(listenTopic, fullTopic, metadata);
        }
        // 不续期
        else if (!renew) {
            final String readableTabKey = RedisKeyHelper.readableKey(this.properties.getSystemName(), this.properties.getIsolation(), listenTopic);

            // 开启了 deny 的持久化与消息非分段消费，才会进入持久化操作
            if (this.properties.isEnableDenyPersistence() && !metadata.isSegment()) {
                final String denyTabKey = RedisKeyHelper.tabKey(this.properties.getSystemName(), SpartaConstant.TABLE_DENY_NAME);
                // 从待消费的队列移动到未确认的队列
                this.denyTabOperate.doProcessDenyPersistence(fullTopic, readableTabKey, denyTabKey);
            }
            // 未开启 deny 保存
            else {
                final String metadataTabKey = RedisKeyHelper.tabKey(this.properties.getSystemName(),
                        SpartaConstant.TABLE_METADATA_NAME);
                this.readableTabOperate.delMetadataReadableTab(fullTopic, readableTabKey, metadataTabKey);
            }

            log.warn("[Sparta] Not ack， topic: {}, metadata: {}", listenTopic, JacksonHelper.serialize(metadata.getBody()));
        }

        if (renew) {
            this.renewFeature.doProcessRenew (fullTopic, listenTopic, metadata, channel);
        }
    }

    /**
     * 执行 ack 逻辑
     *
     * @param listenTopic       @SpartaListener 内中 topics 描述
     * @param fullTopic         完整的 topic：topic:metaId
     * @param metadata          被深拷贝的 body 对象，防止用户在回调方法中进行修改内容
     */
    private void doAcknowledge(String listenTopic, String fullTopic, Metadata<V> metadata) {
        final String readableTabKey = RedisKeyHelper.readableKey(this.properties.getSystemName(), this.properties.getIsolation(), listenTopic);
        final String metadataTabKey = RedisKeyHelper.tabKey(this.properties.getSystemName(),
                SpartaConstant.TABLE_METADATA_NAME);

        // 如果是 segment，则需要判断当前消费的个数
        if (metadata.isSegment()) {
            final String metadataField = RedisKeyHelper.fieldName(listenTopic, metadata.getMetaId());
            this.doSegmentAcknowledge(fullTopic, metadata.getSegmentSize(), readableTabKey, metadataTabKey, metadataField);
            if (log.isDebugEnabled()) {
                log.debug("[Sparta] Acknowledge segment, topic {} call business success, metadata: {}",
                        listenTopic, JacksonHelper.serialize(metadata.getBody()));
            }
            return;
        }

        // ack 后, 删除 metadata_tab 与 readable_tab
        this.readableTabOperate.delMetadataReadableTab(fullTopic, readableTabKey, metadataTabKey);
        if (log.isDebugEnabled()) {
            log.debug("[Sparta] Acknowledge topic: {} call business success, metadata: {}",
                    listenTopic, JacksonHelper.serialize(metadata.getBody()));
        }
    }

    /**
     * 处理分段消费的 ack 操作
     * 因为分段消费中，在 timeout_tab 和 readable_tab 两个队列中的 topic 与 metadata_tab 中
     * 存储的内容是不相同的，在 timeout_tab 和 readable_tab 内容：topic:metaId-segmentNum
     * 而在 metadata_tab 中，则只有 topic:metaId， 所以只有当消费完毕的才会删除 metadata_tab
     *
     * @param fullTopic         topic:metaId-segmentNum
     * @param segmentSize       分段大小，用来判断当前 fullTopic 的数量
     * @param readableTabKey    readable_tab key
     * @param metadataTabKey    metadata_tab key
     * @param metadataField     topic:metaId
     */
    private void doSegmentAcknowledge(final String fullTopic,
                                      final int segmentSize,
                                      final String readableTabKey,
                                      final  String metadataTabKey,
                                      final String metadataField) {

        if (fullTopic.contains(SpartaConstant.LINE)) {
            String[] split = fullTopic.split(SpartaConstant.LINE);
            // 最后一个
            if (Integer.parseInt(split[1]) == (segmentSize - 1)) {
                // 消费完毕后删除 readable_tab 与 metadata_tab
                this.readableTabOperate.delSegmentFinishedConsumption(fullTopic, metadataField, readableTabKey, metadataTabKey);
                return;
            }
            // 只删除 readable_tab
            this.readableTabOperate.delSegmentReadableTab (fullTopic, readableTabKey);
        }
    }

    public static ThreadLocal<Integer> getAckTag() {
        return ACK_TAG;
    }

    public static int removeAck() {
        int ack = ACK_TAG.get();
        ACK_TAG.remove();
        return ack;
    }
}
